package cn.piesat.scanning.business.common.algorithm.dataScan;

import cn.piesat.scanning.dto.*;
import com.alibaba.fastjson.JSON;
import cn.piesat.scanning.business.common.AlgorithmDefineAbstract;
import cn.piesat.scanning.service.*;
import cn.piesat.scanning.utils.FileUtil;
import cn.piesat.scanning.utils.MatchTime;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.io.File;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * 数据入库（磁盘扫描）通用逻辑实现
 * @author xhl
 */
public abstract class DataScanAbstract extends AlgorithmDefineAbstract {

    /** 辅助数据类型 */
    public static final String AUX = "Aux";
    /** 源数据类型 */
    public static final String DIST = "dist";
    /** Kafka方式 */
    public static final String KAFKA = "kafka";

    /**质检源数据类型*/
    public static final String VALIDATION = "validation";

    @Value("${scanning.data.batch.pagesize:1000}")
    private int batchPageSize;

    @Autowired
    private HtPathConfigurationService htPathConfigurationService;
    @Autowired
    private HtScheduleJobService htScheduleJobService;
    @Autowired
    private DbDmsScanningMetadataAuxInfoService dbDmsScanningMetadataAuxInfoService;
    @Autowired
    private HtScanningMetadataService htScanningMetadataService;
    @Autowired
    private DbDmsSchedulerTaskFileRecordService dbDmsSchedulerTaskFileRecordService;
    @Autowired
    private DbDmsScanningValidationService dbDmsScanningValidationService;

    private static final Logger logger = LoggerFactory.getLogger(DataScanAbstract.class);
    /**
     * 默认磁盘扫描插件的执行逻辑，按需使用或重写
     * @param params 插件输入参数
     * @param taskId 任务ID
     */
    @Override
    public void exec(Map<String, Object> params, String taskId) {
        DataScanParams dataScanParams = JSON.parseObject(JSON.toJSONString(params), DataScanParams.class);
        HtScheduleJobDTO taskDTO = htScheduleJobService.findById(taskId);
        HtPathConfigurationDTO pathDTO = htPathConfigurationService.findById(taskDTO.getPathConfigId());
        String dataSource = dataScanParams.getDataSource();
        if (dataSource.equals(KAFKA)) {
            kafkaSave(taskId,pathDTO);
        } else {
            if (dataSource.equals(AUX) || dataSource.equals(DIST) || dataSource.equals(VALIDATION)) {
                //获得所有满足条件的扫描路径
                List<String> dirPathList = processParams(pathDTO.getScanningPath(), dataScanParams);
                List<Object> pojoList = new ArrayList<>();
                for (String path : dirPathList) {
                pojoList.addAll(findAndAssembleData(path,pathDTO,dataScanParams,taskId));
                }
                saveDb(pojoList,dataSource,taskId);
            }
        }
    }

    /**
     * 解析输入参数，获取需要扫描的路径信息集合
     * @param scanPath 扫描路径
     * @param dataScanParams 输入参数对象
     * @return
     */
    public List<String> processParams(String scanPath, DataScanParams dataScanParams) {
        String startTime = dataScanParams.getStartTime();
        String endTime = dataScanParams.getEndTime();
        String lately = dataScanParams.getLately();
        String dirPathRule = dataScanParams.getDirPathRule();
        List<String> dirPath = new ArrayList<>();
        //如果扫描路径为空，则数据扫描自定义算法不继续执行
        if (StringUtils.isNotEmpty(scanPath)) {
            //填入开始时间和最近多少分钟的两种情况，优先填入开始时间参数
            if (StringUtils.isNotEmpty(startTime)) {
                dataScanParams.setStartTime(conventDateTimeFormat(startTime));
                dataScanParams.setEndTime(conventDateTimeFormat(endTime));
                String completePathRule = scanPath + File.separator + dirPathRule;
                if(checkBraces(dirPathRule)) {
                    List<Date> dayList = getTimeZones(dataScanParams.getStartTime(), dataScanParams.getEndTime());
                    List<String> dirList = new ArrayList<>();
                    for (Date date : dayList) {
                        String pathByDate = FileUtil.getPathByDate(completePathRule,date);
                        if (new File(pathByDate).exists()) {
                            dirList.add(pathByDate);
                        }
                    }
                    dirPath = dirList.stream().distinct().collect(Collectors.toList());
                }else{
                    dirPath = new ArrayList<>();
                    dirPath.add(completePathRule);
                }
            } else if (StringUtils.isNotEmpty(lately)) {
                long late = Long.parseLong(lately);
                DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyyMMddHHmm");
                dataScanParams.setStartTime(LocalDateTime.now().minusMinutes(late).format(df).substring(0,8));
                dataScanParams.setEndTime(LocalDateTime.now().format(df).substring(0,8));
                String completePathRule = scanPath + File.separator + dirPathRule;
                if(checkBraces(dirPathRule)) {
                    List<Date> dayList = getTimeZones(dataScanParams.getStartTime(), dataScanParams.getEndTime());
                    List<String> dirList = new ArrayList<>();
                    for (Date date : dayList) {
                        String pathByDate = FileUtil.getPathByDate(completePathRule,date);
                        if (new File(pathByDate).exists()) {
                            dirList.add(pathByDate);
                        }
                    }
                    dirPath = dirList.stream().distinct().collect(Collectors.toList());
                }else{
                    dirPath = new ArrayList<>();
                    dirPath.add(completePathRule);
                }
            }
        }
        return dirPath;
    }

    /**
     * 扫描文件，将符合的文件解析至对象中
     * @return
     */
    public List<Object> findAndAssembleData(String scanDir, HtPathConfigurationDTO pathDTO, DataScanParams dataScanParams,String taskId) {
        List<Object> pojoList = new ArrayList<>();
        //查询数据库中已经存在的文件名
        Set<String> existFileNameSet = new HashSet<>();
        existFileNameSet = dbDmsSchedulerTaskFileRecordService.listByTaskId(taskId);
        //递归组装数据
        recursionFindFile(scanDir,pathDTO,dataScanParams,pojoList,existFileNameSet);
        return pojoList;
    }

    /**
     * 递归遍历文件夹，根据参数将符合的放入对象集合中
     */
    public void recursionFindFile(String scanDir, HtPathConfigurationDTO pathDTO, DataScanParams dataScanParams,
                                  List<Object> pojoList, Set<String> existFileNameSet) {
        logger.info("[磁盘扫描]进入{}目录",scanDir);
        String fileNamePattern = dataScanParams.getFileNamePattern();
        String fileDatePattern = dataScanParams.getFileDatePattern();
        String start = dataScanParams.getStartTime();
        String end = dataScanParams.getEndTime();
        int subLength = start.length();
        long s = Long.parseLong(start);
        long e = Long.parseLong(end);
        logger.info("[磁盘扫描]文件正则={}",fileNamePattern);
        Pattern p = Pattern.compile(fileNamePattern);
        if(!StringUtils.isEmpty(scanDir)) {
            File file = new File(scanDir);
            if (!file.exists()) {
                logger.info("文件不存在{}",file.getPath());
                return;
            }
            File[] listFiles = file.listFiles();
            if (listFiles != null) {
                for (File f : listFiles) {
                    if (f.isDirectory()) {
                        recursionFindFile(f.getAbsolutePath(),pathDTO,dataScanParams,pojoList,existFileNameSet);
                    } else {
                        if(f.exists() && f.isFile()) {
                            String fileName = f.getName();
                            Matcher matcher = p.matcher(fileName);
                            if(matcher.matches()) {
                                String dateTimeStr = getDateTimeStr(fileName,fileDatePattern);
                                long date = Long.parseLong(dateTimeStr.length() > subLength ? dateTimeStr.substring(0,subLength) : dateTimeStr);
                                /** 判断是否在时间范围内 */
                                if(date >= s && date <= e) {
                                    if (dataScanParams.getDataSource().equals(AUX)) {
                                        assembleAux(pojoList,pathDTO,f,existFileNameSet,dataScanParams.getAuxType());
                                    } else if (dataScanParams.getDataSource().equals(DIST) || dataScanParams.getDataSource().equals(VALIDATION)) {
                                        assembleDist(pojoList,pathDTO,f,existFileNameSet,dataScanParams);
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }


    /**
     * 辅助数据信息解析存储至对象中
     * @param pojoList 结果对象集合
     * @param pathDTO 路径对象
     * @param file 当前文件
     * @param existFileNameSet 已存在的文件名名集合
     */
    public abstract void assembleAux(List<Object> pojoList, HtPathConfigurationDTO pathDTO,
                                     File file, Set<String> existFileNameSet, String auxType);

    /**
     * 源数据信息解析存储至对象中
     * @param pojoList 结果对象集合
     * @param pathDTO 路径对象
     * @param file 当前文件
     * @param existFileNameSet 已存在的文件名名集合
     */
    public abstract void assembleDist(List<Object> pojoList, HtPathConfigurationDTO pathDTO,
                                      File file, Set<String> existFileNameSet,DataScanParams dataScanParams);

    /**
     * 将对象存储至数据库
     */
    public void saveDb(List<Object> pojoList, String dataType,String taskId) {
        if (dataType.equals(AUX)) {
            List<DbDmsScanningMetadataAuxInfoDTO> auxInfoList = pojoList.stream()
                    .map(pojo -> (DbDmsScanningMetadataAuxInfoDTO) pojo).collect(Collectors.toList());
            List<String> fileNameList = auxInfoList.stream().
                    map(DbDmsScanningMetadataAuxInfoDTO::getFileName).collect(Collectors.toList());
            dbDmsSchedulerTaskFileRecordService.prepareDataByFileNameList(taskId,fileNameList);
            ExecutorService threadPool = Executors.newFixedThreadPool(30);
            Lists.partition(auxInfoList, batchPageSize)
                    .forEach(subList -> CompletableFuture.runAsync(() -> dbDmsScanningMetadataAuxInfoService.addEntityBatch(subList, taskId), threadPool));
            threadPool.shutdown();
        } else if (dataType.equals(DIST)) {
            List<HtScanningMetadataDTO> dataList = pojoList.stream()
                    .map(pojo -> (HtScanningMetadataDTO) pojo).collect(Collectors.toList());
            List<String> fileNameList = dataList.stream()
                    .map(HtScanningMetadataDTO::getFileName).collect(Collectors.toList());
            dbDmsSchedulerTaskFileRecordService.prepareDataByFileNameList(taskId,fileNameList);
            ExecutorService threadPool = Executors.newFixedThreadPool(30);
            Lists.partition(dataList, batchPageSize)
                    .forEach(subList -> CompletableFuture.runAsync(() -> htScanningMetadataService.addEntityBatch(subList, taskId), threadPool));
            threadPool.shutdown();
        }else if(dataType.equals(VALIDATION)){
            List<DbDmsScanningValidationDTO> dataList = pojoList.stream()
                    .map(pojo -> (DbDmsScanningValidationDTO) pojo).collect(Collectors.toList());
            List<String> fileNameList = dataList.stream()
                    .map(DbDmsScanningValidationDTO::getFileName).collect(Collectors.toList());
            dbDmsSchedulerTaskFileRecordService.prepareDataByFileNameList(taskId,fileNameList);
            ExecutorService threadPool = Executors.newFixedThreadPool(30);
            Lists.partition(dataList, batchPageSize)
                    .forEach(subList -> CompletableFuture.runAsync(() -> dbDmsScanningValidationService.addEntityBatch(subList, taskId), threadPool));
            threadPool.shutdown();
        }
    }

    /**
     * Kafka方式数据入库
     * @param taskId 任务ID
     * @param pathDTO 路径对象
     */
    public void kafkaSave(String taskId, HtPathConfigurationDTO pathDTO) {

    }


    /**
     * 根据正则匹配获取到日期，如果小于12位则补0
     * @param fileName
     * @param dataFilePattern
     * @return
     */
    private String getDateTimeStr(String fileName,String dataFilePattern){
        String str = "";
        if(dataFilePattern != null && !"".equals(dataFilePattern)){
            Pattern p = Pattern.compile(dataFilePattern);
            Matcher m = p.matcher(fileName);
            boolean result = m.find();
            while (result){
                str = m.group(0);
                break;
            }
        }
        if(str.length() < 12){
            int leftlenth = 12 - str.length();
            for(int i = 0 ; i < leftlenth ; i++){
                str += "0";
            }
        }
        return str;
    }

    /**
     * 判断字符串中是否存在 {}
     * @param path
     * @return boolean
     */
    private boolean checkBraces(String path) {
        String leftBraces = "{";
        return path.contains(leftBraces);
    }

    /**
     * 格式化时间
     * @param string
     * @return String datetime
     */
    private String conventDateTimeFormat(String string){
        String dateTime = "";
        DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        DateTimeFormatter df1 = DateTimeFormatter.ofPattern("yyyyMMdd");
        //判断是否为空,空的情况直接获取当前时间 格式 yyyyMMdd
        if(StringUtils.isEmpty(string)) {
            dateTime = LocalDateTime.now().format(df1);
            return dateTime;
        }
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");

        try {
            //格式为 '{yyyy+-}{MM+-}{dd+-}{HH+-}{mm+-}{ss+-}'
            if(checkBraces(string)) {
                Date date = MatchTime.getDate(new Date(), string);
                dateTime = sdf.format(date);
            }else {
                dateTime = LocalDateTime.parse(string, df).format(df1);
            }
        } catch ( Exception e){
            e.printStackTrace();
        }
        return dateTime;
    }

    /**
     * 获取时间区间中所有的天
     * @param startTime
     * @param endTime
     * @return List<Date>
     */
    private List<Date> getTimeZones(String startTime, String endTime){
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
        List<Date> list = new ArrayList<Date>();
        try {
            Date startDate = sdf.parse(startTime);
            Date endDate = sdf.parse(endTime);
            for (long i = (startDate.getTime()-(24*60*60*1000)); i < endDate.getTime(); i++) {
                i += (24*60*60*1000);
                Date date = new Date();
                date.setTime(i);
                list.add(date);
            }
        } catch (ParseException e) {
            e.printStackTrace();
        }
        return list;
    }
}
